#include "pch.hpp"
#include "threadpool_scheduler.hpp"

using namespace services;




threadpool_scheduler::threadpool_scheduler(boost::shared_ptr<const provider> p)
	: scheduler(p)
{
	uint32 maxThreads = boost::thread::hardware_concurrency();
	if(maxThreads == 0)
		maxThreads = 1;

	for(uint32 i = 0; i < maxThreads; ++i)
	{
		m_threads.push_back(boost::make_shared<thread>());
	}

	// Add the actual task that makes this scheduler work.
	// This one simply removes old threads that are no longer needed
	future<void> ft;
	auto fn = [=](progress_listener&){ task_run(); };
	task::ptr t(new task(fn, ft));
	t->setLooped(true);

	m_threads.front()->add(t);
}
///////////////////////////////////////////////////////////////////////////////////////////////////

threadpool_scheduler::~threadpool_scheduler()
{
	// each thread waits in it's destructor for the thread to end
}
///////////////////////////////////////////////////////////////////////////////////////////////////




std::vector<task::const_ptr> threadpool_scheduler::activeTasks()
{
	std::vector<task::const_ptr> t;

	// Go through all threads and have them add their tasks to the vector
	read_lock lock(m_mutex);
	std::for_each(m_threads.begin(), m_threads.end(), [&](const thread::ptr& th)
	{
		auto _tasks = th->activeTasks();
		t.insert(t.end(), _tasks.begin(), _tasks.end());
	});

	return t;
}
///////////////////////////////////////////////////////////////////////////////////////////////////

void threadpool_scheduler::executeTask(task::ptr t)
{
	if(t->dedicated())
	{
		// The task wants to run dedicated: we grant the wish
		// as long as the os is willing to spawn new threads
		write_lock lock(m_mutex);
		m_threads.push_back(boost::make_shared<thread>(t));
	}
	else
	{
		// The task doesn't need a dedicated thread, add it to one of the existing threads
		thread::ptr thread_to_add;
		read_lock lock(m_mutex);

		// We simply look for the thread with the least number of tasks
		std::for_each(m_threads.begin(), m_threads.end(), [&](const thread::ptr& t)
		{
			if(thread_to_add == nullptr || (t->numTasks() > thread_to_add->numTasks() && !t->exclusive()))
				thread_to_add = t;
		});

		thread_to_add->add(t);
	}
}
///////////////////////////////////////////////////////////////////////////////////////////////////




void threadpool_scheduler::task_run()
{
	//
	// Remove old threads
	//

	write_lock lock(m_mutex, boost::try_to_lock);
	if(lock.owns_lock())
	{
		for(auto i = m_threads.begin(); i != m_threads.end();)
		{
			// Exclusive threads without tasks are removed
			if((*i)->exclusive() && (*i)->numTasks() == 0)
				i = m_threads.erase(i);
			else
				++i;
		}
	}
}
///////////////////////////////////////////////////////////////////////////////////////////////////
